package com.example.pro2_reactive_streams;

import java.io.IOException;
import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

/**
 * @program: reactivePro
 * @description:
 * @author: zy
 * @create: 2025-04-17 18:10
 */
public class Pro1_Publisher_subscriber {
    //发布者publisher -> 处理器p1  ->  订阅者s1
    public static void main(String[] args) {
        //Processor: 处理器，同时也是发布者publisher和订阅者scriber
        Flow.Processor<Character, String> p1 = new Flow.Processor<>() {
            //这个处理器的下一级 订阅者
            private Flow.Subscriber<? super String> subscriber;

            //继承自 publisher
            @Override
            public void subscribe(Flow.Subscriber<? super String> subscriber) {
                this.subscriber = subscriber;
            }

            //继承自 subscriber, onSubscribe() 方法作用是
            // Subscription 是 Publisher 和 Subscriber 之间的“桥梁”，负责管理两件事：
            //1. 背压（Backpressure）：控制 Subscriber 想要接收多少数据
            //2. 取消订阅：Subscriber 可以随时取消接收数据。
            //它里面有两个方法:
            //1. request(long n)：请求 Publisher 发布 n 个数据项。
            //2. cancel()：取消订阅，Publisher 停止发送数据。
            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                subscriber.onSubscribe(subscription);
            }

            @Override
            public void onNext(Character item) {
                //对上一级发布者发布的数据进行处理
                String s=order(item);
                subscriber.onNext(   s   );
            }

            @Override
            public void onError(Throwable throwable) {
                subscriber.onError(throwable);
            }

            @Override
            public void onComplete() {
                subscriber.onComplete();
            }

            private String order(Character c) {
                return "Char is: " + c;
            }
        };
        // 订阅者
        Flow.Subscriber<String> s1 = new Flow.Subscriber<>() {
            private Flow.Subscription subscription;  //接收上一级发布者的订阅信息
            @Override // 订阅者订阅发布者时，调用的方法,
            public void onSubscribe(Flow.Subscription subscription) {
                this.subscription = subscription;
                subscription.request(1);   //请求一个数据, 这是一个背压的操作， 表示订阅者可以控制速率.
            }
            @Override
            public void onNext(String item) {
                //作为订阅者，对发布者发布的数据进行消费
                store(item);
                subscription.request(1);  //背压，请求一个数据
            }
            @Override
            public void onError(Throwable throwable) {
                System.out.println("=========throwable=========" + throwable);
            }
            @Override
            public void onComplete() {
                System.out.println("=========onComplete=========");
            }
            private void store(String s) {
                new Thread(()->{
                    //业务方法
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                    System.out.println("Received: " + s);
                }).start();

            }
        };

        //配置以形成 pipeline:    发布者publisher -> 处理器p1  ->  订阅者s1
        // 发布者订阅订阅者
        SubmissionPublisher<Character> publisher = new SubmissionPublisher<>();
        publisher.subscribe(p1);    // 发布者订阅处理器p1
        p1.subscribe(s1);           // 处理器p1订阅订阅者s1
        //所以流程为:  发布者publisher -> 处理器p1  ->  订阅者s1
        for (int i = 0; i < 10; i++) {
            //发布者发布数据:   A B C D E F G H I J
            publisher.submit((char) ('A' + i));
        }
        // 让主线程在此等待
        try {
            System.out.println("begin");
            System.in.read();
            System.out.println("end");
        } catch (IOException e) {
            e.printStackTrace();
        }
        publisher.close();
    }


}
